package com.booter.zookeeper;

import com.booter.zookeeper.listener.ChildListener;
import com.booter.zookeeper.listener.ConnStateListener;
import com.booter.zookeeper.listener.DataListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @param <T> TargetChildListener
 * @param <K> TargetDataListener
 * @author admin
 */
public abstract class AbstractZkClient<T, K> implements ZkClient {

    private static final Logger logger = LoggerFactory.getLogger(AbstractZkClient.class);
    private final Set<ConnStateListener> stateListeners = new CopyOnWriteArraySet<>();
    //String, ConcurrentMap<ChildListener, T>
    private final ConcurrentMap<String/* path */, ConcurrentMap<ChildListener, T>> childListeners = new ConcurrentHashMap<>();

    // String, ConcurrentMap<DataListener, K>
    private final ConcurrentMap<String/* path */, ConcurrentMap<DataListener, K>> dataListeners = new ConcurrentHashMap<>();

    private boolean closed;

    @Override
    public String create(String path, boolean ephemeral, boolean sequential) {
        int i = path.lastIndexOf('/');
        if (i > 0) {
            this.create(path.substring(0, i), false, false);
        }
        if (ephemeral) {
            return this.createEphemeral(path, sequential);
        } else {
            return this.createPersistent(path, sequential);
        }
    }

    @Override
    public String create(String path, Object data, boolean ephemeral,
                         boolean sequential) {
        int i = path.lastIndexOf('/');
        if (i > 0) {
            this.create(path.substring(0, i), data, false, false);
        }
        if (ephemeral) {
            return this.createEphemeral(path, data, sequential);
        } else {
            return this.createPersistent(path, data, sequential);
        }
    }

    public Set<ConnStateListener> getSessionListeners() {
        return this.stateListeners;
    }

    @Override
    public void addConnStateListener(ConnStateListener listener) {
        this.stateListeners.add(listener);
    }

    @Override
    public void removeStateListener(ConnStateListener listener) {
        this.stateListeners.remove(listener);
    }

    @Override
    public List<String> addChildListener(String path, final ChildListener listener) {
        ConcurrentMap<ChildListener, T> listeners = this.childListeners.get(path);
        if (listeners == null) {
            this.childListeners.putIfAbsent(path, new ConcurrentHashMap<>());
            listeners = this.childListeners.get(path);
        }
        T targetListener = listeners.get(listener);
        if (targetListener == null) {
            listeners.putIfAbsent(listener, this.createTargetChildListener(path, listener));
            targetListener = listeners.get(listener);
        }
        return this.addTargetChildListener(path, targetListener);
    }

    @Override
    public void removeChildListener(String path, ChildListener listener) {
        ConcurrentMap<ChildListener, T> listeners = this.childListeners.get(path);
        if (listeners != null) {
            T targetListener = listeners.remove(listener);
            if (targetListener != null) {
                this.removeTargetChildListener(path, targetListener);
            }
        }
    }

    @Override
    public void addDataListener(String path, DataListener listener) {
        ConcurrentMap<DataListener, K> listeners = this.dataListeners.get(path);
        if (listeners == null) {
            this.dataListeners.putIfAbsent(path, new ConcurrentHashMap<>());
            listeners = this.dataListeners.get(path);
        }
        K targetListener = listeners.get(listener);
        if (targetListener == null) {
            listeners.putIfAbsent(listener, this.createTargetDataListener(path, listener));
            targetListener = listeners.get(listener);
        }
        this.addTargetDataListener(path, targetListener);
    }

    @Override
    public void removeDataListener(String path, DataListener listener) {
        ConcurrentMap<DataListener, K> listeners = this.dataListeners.get(path);
        if (listeners != null) {
            K targetListener = listeners.remove(listener);
            if (targetListener != null) {
                this.removeTargetDataListener(path, targetListener);
            }
        }
    }


    private void removeAllDataListener() {
        dataListeners.forEach((k, v) -> v.forEach((x, y) -> this.removeDataListener(k, x)));
        dataListeners.clear();
    }


    private void removeAllChildListener() {
        childListeners.forEach((k, v) -> v.forEach((x, y) -> this.removeTargetChildListener(k, y)));
        childListeners.clear();
    }

    @Override
    public void close() {
        if (this.closed) {
            return;
        }
        try {
            removeAllDataListener();
            removeAllChildListener();
            this.closed = true;
            this.doClose();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    protected void stateChanged(int state) {
        for (ConnStateListener stateListener : this.getSessionListeners()) {
            stateListener.stateChanged(state);
        }
    }

    protected abstract void doClose();

    /***
     * 创建永久的自增序列路径
     *
     * @param path
     * @param sequential
     *        是否为递增序列
     * @return
     */
    protected abstract String createPersistent(String path, boolean sequential);

    /**
     * 创建永久的自增序列路径 并设置节点数据
     *
     * @param path       路径
     * @param data       已序列化数据
     * @param sequential 是否为递增序列
     * @return
     */
    protected abstract String createPersistent(String path, Object data,
                                               boolean sequential);

    /**
     * 创建临时自增序列路径
     *
     * @param path       路径
     * @param sequential 是否为递增序列
     * @return
     */
    protected abstract String createEphemeral(String path, boolean sequential);

    /**
     * 创建临时自增序列路径 并设置节点数据
     *
     * @param path       路径
     * @param data       已序列化数据
     * @param sequential 是否为递增序列
     * @return
     */
    protected abstract String createEphemeral(String path, Object data,
                                              boolean sequential);

    /**
     * 创建节点监听器
     *
     * @param path
     * @param listener
     * @return
     */
    protected abstract T createTargetChildListener(String path,
                                                   ChildListener listener);

    protected abstract List<String> addTargetChildListener(String path,
                                                           T listener);

    protected abstract void removeTargetChildListener(String path, T listener);

    protected abstract void addTargetDataListener(String path,
                                                  K targetListener);

    protected abstract K createTargetDataListener(String path,
                                                  DataListener listener);

    protected abstract void removeTargetDataListener(String path,
                                                     K targetListener);

}
